package com.millstein.flume.interceptor;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.stream.Collectors;

/**
 * 解决时间漂移问题的拦截器，将flume中的时间戳改写为事件时间
 */
public class TimeInterceptor implements Interceptor {
    @Override
    public void initialize() {

    }

    @Override
    public Event intercept(Event event) {
        // 1.将event中的内容取出，并转为json对象
        JSONObject json = JSON.parseObject(new String(event.getBody(), StandardCharsets.UTF_8));
        // 2.从json对象中取出事件时间（字段参考业务数据采集章节中的内容）
        Long eventTime = json.getLong("ts");
        // 3.将事件时间写入到事件中的头部信息中（字段参考官网）
        event.getHeaders().put("timestamp", eventTime.toString());
        // 4.返回事件
        return event;
    }

    @Override
    public List<Event> intercept(List<Event> list) {
        for (Event event : list) {
            intercept(event);
        }
        return list;
    }

    @Override
    public void close() {

    }

    public static class Builder implements Interceptor.Builder {

        @Override
        public Interceptor build() {
            return new TimeInterceptor();
        }

        @Override
        public void configure(Context context) {

        }
    }
}
